//package com.zhz.mall.portal.kakfa;
//
//import cn.hutool.core.lang.Assert;
//import cn.hutool.json.JSONUtil;
//import com.alibaba.fastjson.JSON;
//import com.zhz.mall.common.util.validate.ValidationUtils;
//import lombok.SneakyThrows;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.context.ApplicationContext;
//import org.springframework.core.BridgeMethodResolver;
//import org.springframework.util.ReflectionUtils;
//import org.springframework.validation.annotation.Validated;
//
//import java.lang.reflect.InvocationTargetException;
//import java.lang.reflect.Method;
//import java.lang.reflect.Parameter;
//import java.lang.reflect.UndeclaredThrowableException;
//import java.util.Objects;
//
///**
// * DefaultKafkaMessageListener
// * <p>
// * 默认 kafka 消息监听器
// *
// * @author zhouhengzhe
// */
//@Slf4j
//public class DefaultKafkaMessageListener implements KafkaMessageListener {
//
//    private final ApplicationContext applicationContext;
//
//    private final String beanName;
//    private final Class targetClass;
//    private final String name;
//    private final Method method;
//    private final String topic;
//    private final boolean valid;
//    private final Class<?> payloadType;
//    private Object targetBean;
//
//
//    DefaultKafkaMessageListener(ApplicationContext context, String beanName, Class targetClass, Method method, KafkaListener ann) {
//        this.applicationContext = context;
//        this.beanName = beanName;
//        this.targetClass = targetClass;
//        this.method = BridgeMethodResolver.findBridgedMethod(method);
//        this.name = targetClass.getName() + '.' + method.getName();
//        this.topic = Assert.notBlank(ann.topic());
//        Class[] parameterTypes = this.method.getParameterTypes();
//        Assert.state(parameterTypes.length == 1,
//                "the @KafkaListener method must only contains one parameter on bean '{}': {}", beanName, this.method);
//        this.payloadType = parameterTypes[0];
//        Parameter parameter = this.method.getParameters()[0];
//        this.valid = ann.valid() || parameter.isAnnotationPresent(Validated.class);
//        ReflectionUtils.makeAccessible(this.method);
//    }
//
//
//    @Override
//    public String getName() {
//        return this.name;
//    }
//
//
//    @Override
//    public String getTopic() {
//        return this.topic;
//    }
//
//
//    @Override
//    public void onMessage(ConsumerRecord record) {
//        Assert.state(this.topic.equals(record.topic()));
//        String value = record.value().toString();
//        Assert.notBlank(value, "Kafka 消息不能为空");
//        Object payload = doParse(value);
//        doValid(payload);
//        doInvoke(payload);
//    }
//
//
//    /**
//     * 数据解析
//     */
//    private Object doParse(String value) {
//        // 如果接收类型为 String 则不处理
//        if (this.payloadType == String.class) {
//            return value;
//        }
//
//        if (!JSONUtil.isTypeJSONObject(value)) {
//            throw new RuntimeException("非法 Kafka 消息, 必须为 json 格式");
//        }
//
//        return JSON.parseObject(value, this.payloadType);
//    }
//
//
//    /**
//     * 数据校验
//     */
//    private void doValid(Object payload) {
//        if (payload instanceof String) {
//            return;
//        }
//        if (this.valid) {
//            ValidationUtils.valid(payload);
//        }
//    }
//
//
//    /**
//     * 消息处理
//     */
//    @SneakyThrows
//    private void doInvoke(Object payload) {
//        Object bean = getTargetBean();
//
//        try {
//            this.method.invoke(bean, payload);
//        } catch (InvocationTargetException ex) {
//            Throwable targetException = ex.getTargetException();
//            if (targetException instanceof RuntimeException) {
//                throw targetException;
//            } else {
//                throw new UndeclaredThrowableException(targetException, "Failed to invoke event listener method");
//            }
//        }
//    }
//
//
//    /**
//     * 获取目标 bean
//     */
//    private Object getTargetBean() {
//        if (Objects.isNull(this.targetBean)) {
//            this.targetBean = this.applicationContext.getBean(this.beanName);
//        }
//        return this.targetBean;
//    }
//
//
//}
